package com.boarsoft.message.core;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.boarsoft.bean.ReplyInfo;
import com.boarsoft.common.util.JsonUtil;
import com.boarsoft.message.bean.MessageEntity;
import com.boarsoft.message.bean.MessageQueryParam;
import com.boarsoft.message.bean.MessageQueueStat;
import com.boarsoft.message.bean.RecipientMQ;
import com.boarsoft.message.bean.RecipientQueueInfo;
import com.boarsoft.message.bean.TargetMQ;
import com.boarsoft.message.bean.TargetQueueInfo;
import com.boarsoft.message.broker.MessageBroker;
import com.boarsoft.message.store.MessageRoller;

/**
 * 供message-web调用，以动态创建、更新、删除、查询队列
 * 
 * @author Mac_J
 *
 */
public class MessageBrokerImpl implements MessageBroker {
	private static final Logger log = LoggerFactory.getLogger(MessageBrokerImpl.class);

	@Autowired
	protected MessageContext messageContext;

	@Override
	public ReplyInfo<Object> activeTargetQueue(String key) {
		return messageContext.activeTargetQueue(key);
	}

	@Override
	public ReplyInfo<Object> deactiveTargetQueue(String key) {
		return messageContext.deactiveTargetQueue(key);
	}

	@Override
	public ReplyInfo<Object> activeRecipientQueue(String key) {
		return messageContext.activeRecipientQueue(key);
	}

	@Override
	public ReplyInfo<Object> deactiveRecipientQueue(String key) {
		return messageContext.deactiveRecipientQueue(key);
	}

	@Override
	public ReplyInfo<Object> removeTargetQueue(String key, boolean force) {
		return messageContext.removeTargetQueue(key, force);
	}

	@Override
	public ReplyInfo<Object> removeRecipientQueue(String key, boolean force) {
		return messageContext.removeRecipientQueue(key, force);
	}

	@Override
	public ReplyInfo<Object> addTargetQueue(String key, TargetMQ tq) {
		return messageContext.addTargetQueue(key, tq);
	}

	@Override
	public ReplyInfo<Object> addTargetQueue(String key, String xml) {
		return messageContext.addTargetQueue(key, xml);
	}

	@Override
	public ReplyInfo<Object> updateTargetQueue(String key, String xml) {
		return messageContext.updateTargetQueue(key, xml);
	}

	@Override
	public ReplyInfo<Object> addRecipientQueue(String key, String xml) {
		return messageContext.addRecipientQueue(key, xml);
	}

	@Override
	public ReplyInfo<Object> updateRecipientQueue(String key, String xml) {
		return messageContext.updateRecipientQueue(key, xml);
	}

	@Override
	public List<MessageEntity> query(String target, MessageQueryParam mqp) {
		List<MessageEntity> lt = new LinkedList<MessageEntity>();
		TargetMQ tq = messageContext.getTargetQueue(target);
		if (tq == null) {
			log.error("TargetMQ {} not found", tq);
			return lt;
		}
		MessageRoller roller = tq.getRoller();
		if (roller != null) {
			return roller.query(mqp);
		}
		// TODO 从现有队列中提取？
		return lt;
	}

	@Override
	public List<Map<String, Object>> count(String target, MessageQueryParam mqp) {
		List<Map<String, Object>> lt = new ArrayList<>();
		TargetMQ tq = messageContext.getTargetQueue(target);
		if (tq == null) {
			log.error("TargetMQ {} not found", tq);
			return lt;
		}
		MessageRoller roller = tq.getRoller();
		if (roller != null) {
			return roller.count(mqp);
		}
		// TODO 直接从内存队列中统计？
		return lt;
	}

	@Override
	public Map<String, TargetQueueInfo> getTargetQueue() {
		Map<String, TargetQueueInfo> rm = new HashMap<String, TargetQueueInfo>();
		Map<String, TargetMQ> tqMap = messageContext.getTargetQueueMap();
		for (Entry<String, TargetMQ> en : tqMap.entrySet()) {
			TargetMQ q = en.getValue();
			rm.put(en.getKey(), q.getQueueInfo());
		}
		return rm;
	}

	@Override
	public Object statTargetQueue() {
		return this.getTargetQueue();
	}

	@Override
	public MessageQueueStat statTargetQueue(String target) {
		MessageQueueStat o = new MessageQueueStat();
		TargetMQ tq = messageContext.getTargetQueue(target);
		o.setSize(tq.getSize());
		o.setIn(tq.getIn());
		o.setOut(tq.getOut());
		log.info("Message queue {}: {}", target, JsonUtil.from(o));
		return o;
	}

	@Override
	public Map<String, RecipientQueueInfo> statRecipientQueue() {
		Map<String, RecipientQueueInfo> rm = new HashMap<String, RecipientQueueInfo>();
		Map<String, RecipientMQ> rqMap = messageContext.getRecipientQueueMap();
		for (Entry<String, RecipientMQ> en : rqMap.entrySet()) {
			RecipientMQ q = en.getValue();
			rm.put(en.getKey(), q.getQueueInfo());
		}
		return rm;
	}

	@Override
	public MessageQueueStat statRecipientQueue(String recipient) {
		MessageQueueStat o = new MessageQueueStat();
		RecipientMQ rq = messageContext.getRecipientQueue(recipient);
		o.setSize(rq.getSize());
		o.setIn(rq.getIn());
		o.setOut(rq.getOut());
		return o;
	}

	public MessageContext getMessageContext() {
		return messageContext;
	}

	public void setMessageContext(MessageContext messageContext) {
		this.messageContext = messageContext;
	}
}